package com.to8to.kitt.esb;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;

import java.util.List;

import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TMessageMeta;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.TNettyTransport;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;

public class ThriftBusMessageCodec extends ByteToMessageCodec<BusMessage>{

	public static Logger logger = LoggerFactory.getLogger(ThriftBusMessageCodec.class);  

	private TProtocolFactory protocolFactory;
	
	public ThriftBusMessageCodec(TProtocolFactory protocolFactory) {
		this.protocolFactory = protocolFactory;
	}
	
	@Override
	protected void encode(ChannelHandlerContext ctx, BusMessage msg,
			ByteBuf out) throws Exception {
		
		TMessageMeta meta = msg.getMeta();
		
		if(meta == null)
			meta = new TMessageMeta(msg.getHead().name);
		
		LogUtils.openTag(meta.tag);
		
		logger.trace("encode start! seqid: {}", msg.getHead().seqid);
		
		long t1 = System.currentTimeMillis();
		
		out.writeBytes(msg.getBuff());
		
		TTransport transport = new TNettyTransport(ctx.channel(), out);
		TProtocol protocol = protocolFactory.getProtocol(transport);
		try {
			int writerIndex = out.writerIndex();
			
			out.writerIndex(0);
			
			protocol.writeMessageBegin(msg.getHead());
			
			out.writerIndex(writerIndex);
			
			protocol.getTransport().flush();
			
		} catch (Exception e) {
			logger.error( "", e);
		}
		
		long t2 = System.currentTimeMillis();
		
		logger.trace("encode finish! time: {} ms, seqid:{}, head:{}", (t2-t1), msg.getHead().seqid, msg.getHead());
		
		LogUtils.removeTag();
	}


	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in,
			List<Object> out) throws Exception {
		
		logger.trace("decode start: " + in);
		
		long t1 = System.currentTimeMillis();
		
		TTransport transport = new TNettyTransport(ctx.channel(), in);
		
		TProtocol protocol = protocolFactory.getProtocol(transport);
		
		BusMessage msg = new BusMessage();
		
		try {
			int totalReadable = in.readableBytes();
			
			TMessage head = protocol.readMessageBegin();
			
			TMessageMeta meta = head.getTMessageMeta();
			
			if(meta == null)
				meta = new TMessageMeta(msg.getHead().name);
			
			LogUtils.openTag(meta.tag);
			
			msg.setHead(head);
			
			msg.setMeta(meta);
			
			in.readerIndex(0);
			
			msg.setBuff(in.copy());
	
//			msg.setBuff(in.slice());
//			ReferenceCountUtil.retain(in);
			
			out.add(msg);
			
			in.readerIndex(totalReadable);
			
		} catch (Exception e) {
			logger.error(LogUtils.logStackTrace(e));
		}
		
		long t2 = System.currentTimeMillis();
		
		logger.trace("decode finish! time:{} ms, seqid:{}, head:{}", (t2-t1), msg.getSeqid(), msg.getHead());
		
		LogUtils.removeTag();
	}
	
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    	NettyUtils.exitIfOutOfMemory(cause);
        ctx.fireExceptionCaught(cause);
    }
}
